package com.xiguanlezz.cn.common.notify;

import com.xiguanlezz.cn.common.notify.listener.Subscriber;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ClassUtils;

import java.util.Map;
import java.util.NoSuchElementException;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * @Author：习惯了种种
 * @CreateDate：2024/11/28 20:10
 * @Description：事件通知中心
 */
@Slf4j
public class NotifyCenter {
    // 单例
    private static final NotifyCenter INSTANCE = new NotifyCenter();
    // key是事件名称，value就是对应的发布器，这个map存放的是DefaultPublisher事件发布器
    private final ConcurrentMap<String, EventPublisher> publisherMap = new ConcurrentHashMap<>(16);

    // 事件发布器内部的队列定义长度
    public static int ringBufferSize;
    // 事件发布器的类型，这个是要从SPI文件中加载的，如果用户没有设置，就是用默认的DefaultPublisher事件发布器
    private static Class<? extends EventPublisher> clazz;
    // 事件发布器的工厂，这个工厂专门创建事件发布器
    private static final EventPublisherFactory DEFAULT_PUBLISHER_FACTORY;

    private static final AtomicBoolean CLOSED = new AtomicBoolean(false);


    static {
        String ringBufferSizeProperty = "nacos.core.notify.ring-buffer-size";
        ringBufferSize = Integer.getInteger(ringBufferSizeProperty, 16384);

        // 这里用默认的事件发布者，不搞SPI了
        clazz = DefaultPublisher.class;

        // cls就是事件发布器关注的类型，buffer就是事件发布器中事件队列的大小
        DEFAULT_PUBLISHER_FACTORY = (cls, buffer) -> {
            try {
                // 每个事件类型就要定义一个对应的事件发布器
                EventPublisher publisher = clazz.newInstance();
                //事件发布器初始化，在init方法中，事件发布器就会启动了
                publisher.init(cls, buffer);
                return publisher;
            } catch (Throwable throwable) {
                log.error("Service class newInstance has error : ", throwable);
                throw new RuntimeException(throwable);
            }
        };

        // 添加一个JVM钩子函数，当程序关闭的时候，回调这个停止所有事件发布器工作，每个事件发布器都是一个线程
        Runtime.getRuntime().addShutdownHook(new Thread(
                new Runnable() {
                    @Override
                    public void run() {
                        NotifyCenter.shutdown();
                    }
                }
        ));
    }

    /**
     * 根据事件类型创建事件生产者
     */
    public static EventPublisher registerToPublisher(final Class<? extends Event> eventType, final int queueMaxSize) {
        return registerToPublisher(eventType, DEFAULT_PUBLISHER_FACTORY, queueMaxSize);
    }

    public static EventPublisher registerToPublisher(final Class<? extends Event> eventType,
                                                     final EventPublisherFactory factory,
                                                     final int queueMaxSize) {
        // 根据订阅者订阅的事件类型获得一个topic，其实就是订阅事件的字符串
        final String topic = ClassUtils.getCanonicalName(eventType);
        synchronized (NotifyCenter.class) {
            // 判断publisherMap中是否存在了对应的事件发布器，如果不存在，就以topic为key，事件工厂创建的事件发布器为publisherMap的value
            INSTANCE.publisherMap.computeIfAbsent(topic, (innerTopic) -> {
                return factory.apply(eventType, queueMaxSize);
            });
        }
        return INSTANCE.publisherMap.get(topic);
    }

    /**
     * 根据事件类型移除事件生产者
     */
    public static void deregisterPublisher(final Class<? extends Event> eventType) {
        // 根据订阅者订阅的事件类型获得一个topic，其实就是订阅事件的字符串
        final String topic = ClassUtils.getCanonicalName(eventType);
        EventPublisher publisher = INSTANCE.publisherMap.remove(topic);
        try {
            publisher.shutdown();
        } catch (Throwable ex) {
            log.error("There was an exception when publisher shutdown : ", ex);
        }
    }

    /**
     * 添加订阅者到事件通知中心的方法1
     */
    public static void registerSubscriber(Subscriber consumer) {
        registerSubscriber(consumer, DEFAULT_PUBLISHER_FACTORY);
    }

    /**
     * 添加订阅者到事件通知中心的方法2
     */
    public static void registerSubscriber(Subscriber consumer, EventPublisherFactory factory) {
        // 获得订阅者订阅的事件
        Class<? extends Event> subscribeType = consumer.subscribeType();
        // 添加到publisherMap中
        addSubscriber(consumer, subscribeType, factory);
    }

    /**
     * 1. 先根据事件类型判断是否需要新建事件生产者
     * 2. 获取到对应类型的事件生产者后，将事件订阅着注册到生产者上
     */
    private static void addSubscriber(Subscriber consumer, Class<? extends Event> subscribeType,
                                      EventPublisherFactory factory) {
        // 根据订阅者订阅的事件类型获得一个topic，其实就是订阅事件的字符串
        String topic = ClassUtils.getCanonicalName(subscribeType);
        synchronized (NotifyCenter.class) {
            // 判断publisherMap中是否存在了对应的事件发布器，如果不存在，就以topic为key，事件工厂创建的事件发布器为publisherMap的value
            INSTANCE.publisherMap.computeIfAbsent(topic, (innerTopic) -> {
                return factory.apply(subscribeType, ringBufferSize);
            });
        }

        // 得到事件对应的事件发布器
        EventPublisher publisher = INSTANCE.publisherMap.get(topic);
        // 把订阅者添加到事件发布器中
        publisher.addSubscriber(consumer);
    }

    /**
     * 根据事件类型从事件通知中心中移除订阅者
     */
    public static void deregisterSubscriber(final Subscriber consumer) {
        final Class<? extends Event> subscribeType = consumer.subscribeType();
        if (removeSubscriber(consumer, subscribeType)) {
            return;
        }
        throw new NoSuchElementException("The subscriber has no event publisher");
    }

    private static boolean removeSubscriber(final Subscriber consumer, Class<? extends Event> subscribeType) {
        // 根据订阅者订阅的事件类型获得一个topic，其实就是订阅事件的字符串
        final String topic = ClassUtils.getCanonicalName(subscribeType);
        EventPublisher eventPublisher = INSTANCE.publisherMap.get(topic);
        if (Objects.isNull(eventPublisher)) {
            return false;
        }
        eventPublisher.removeSubscriber(consumer);
        return true;
    }

    public static void shutdown() {
        if (!CLOSED.compareAndSet(false, true)) {
            return;
        }
        log.warn("[NotifyCenter] Start destroying Publisher");

        // 遍历关闭所有的事件生产者
        for (Map.Entry<String, EventPublisher> entry : INSTANCE.publisherMap.entrySet()) {
            try {
                EventPublisher eventPublisher = entry.getValue();
                eventPublisher.shutdown();
            } catch (Throwable e) {
                log.error("[EventPublisher] shutdown has error : ", e);
            }
        }
    }

}
